/**
 * Copyright 2016 vip.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 * </p>
 **/

package spring.cloud.tasks.api.service.impl;

import com.google.common.base.Splitter;
import com.google.common.base.Strings;
import com.google.gson.Gson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import spring.cloud.tasks.api.database.entity.TaskConfigurationDto;
import spring.cloud.tasks.api.database.service.TaskConfigurationDtoService;
import spring.cloud.tasks.api.domain.*;
import spring.cloud.tasks.api.service.ReportAlarmService;
import spring.cloud.tasks.api.service.RestApiService;
import spring.cloud.tasks.api.service.TaskService;
import spring.cloud.tasks.api.service.ZooKeeperService;
import spring.cloud.tasks.api.service.helper.ReuseCallBackWithoutReturn;
import spring.cloud.tasks.api.service.helper.ReuseUtils;
import spring.cloud.tasks.api.service.helper.ZooKeeperReUseCallBack;
import spring.cloud.tasks.api.utils.SaturnConstants;
import spring.cloud.tasks.api.utils.TaskNodePath;
import spring.cloud.tasks.api.vo.AlarmInfo;
import spring.cloud.tasks.api.zookeeper.ZooKeeper;
import spring.cloud.tasks.api.zookeeper.ZooKeeperRepository;

import javax.annotation.Resource;
import java.util.*;

/**
 * @author hebelala
 */
public class RestApiServiceImpl implements RestApiService {

    private static final Logger log = LoggerFactory.getLogger(RestApiServiceImpl.class);

    private static final long STATUS_UPDATE_FORBIDDEN_INTERVAL_IN_MILL_SECONDS = 3 * 1000L;

    private static final long OPERATION_FORBIDDEN_INTERVAL_AFTER_CREATION_IN_MILL_SECONDS = 10 * 1000L;

    private static final String JOB_STATUS_NOT_CORRECT_TEMPATE = "job's status is not {%s}";

    private static final String ALL_EXECUTORS_ARE_OFFLINE = "all executors are offline";

    private static final String NO_EXECUTOR_FOUND = "no executor found for this job";

    private static final String EXECUTOR_RESTART_TIME_PREFIX = "restart on ";
    private final Gson gson = new Gson();
    @Resource
    private ZooKeeperService zooKeeperService;
    @Resource
    private ZooKeeperRepository zooKeeperRepository;
    @Resource
    private TaskService taskService;
    @Resource
    private TaskConfigurationDtoService taskConfigurationDtoService;
    @Resource
    private ReportAlarmService reportAlarmService;

    @Override
    public void createJob(final String namespace, final TaskConfiguration taskConfiguration) throws RuntimeException {
        ReuseUtils.reuse(namespace, zooKeeperService, zooKeeperRepository, new ReuseCallBackWithoutReturn() {
            @Override
            public void call(ZooKeeper zooKeeper) throws RuntimeException {
                taskService.addJob(namespace, taskConfiguration, "");
            }
        });

    }

    @Override
    public RestApiJobInfo getRestAPIJobInfo(String namespace, final String jobName) throws RuntimeException {
        return ReuseUtils.reuse(namespace, jobName, zooKeeperService, zooKeeperRepository,
                new ZooKeeperReUseCallBack<RestApiJobInfo>() {
                    @Override
                    public RestApiJobInfo call(ZooKeeper zooKeeper)
                            throws RuntimeException {
                        if (!zooKeeper.checkExists(TaskNodePath.getConfigNodePath(jobName))) {
                            throw new RuntimeException(
                                    "The jobName does not existed");
                        }

                        return constructJobInfo(zooKeeper, jobName);
                    }
                });
    }

    @Override
    public List<RestApiJobInfo> getRestApiJobInfos(final String namespace) throws RuntimeException {
        return ReuseUtils
                .reuse(namespace, zooKeeperService, zooKeeperRepository, new ZooKeeperReUseCallBack<List<RestApiJobInfo>>() {
                    @Override
                    public List<RestApiJobInfo> call(ZooKeeper zooKeeper)
                            throws RuntimeException {
                        List<RestApiJobInfo> restApiJobInfos = new ArrayList<>();
                        List<TaskConfiguration> taskConfigurationList = taskService.getTaskConfigurationList(namespace);
                        if (taskConfigurationList != null) {
                            for (TaskConfiguration taskConfiguration : taskConfigurationList) {
                                try {
                                    RestApiJobInfo restApiJobInfo = constructJobInfo(zooKeeper,
                                            taskConfiguration.getTaskId());
                                    restApiJobInfos.add(restApiJobInfo);
                                } catch (Exception e) {
                                    log.error("getRestApiJobInfos exception:", e);
                                    continue;
                                }
                            }
                        }
                        return restApiJobInfos;
                    }
                });
    }

    private RestApiJobInfo constructJobInfo(ZooKeeper zooKeeper, String job) {
        RestApiJobInfo restApiJobInfo = new RestApiJobInfo();
        restApiJobInfo.setJobName(job);
        // 设置作业配置信息
        setJobConfig(zooKeeper, restApiJobInfo, job);
        // 设置运行状态
        setRunningStatus(zooKeeper, restApiJobInfo, job);
        // 设置统计信息
        RestApiJobStatistics restApiJobStatistics = new RestApiJobStatistics();
        setStatics(zooKeeper, restApiJobStatistics, job);
        restApiJobInfo.setStatistics(restApiJobStatistics);
        return restApiJobInfo;
    }

    private void setRunningStatus(ZooKeeper zooKeeper,
                                  RestApiJobInfo restApiJobInfo, String jobName) {
        String executionNodePath = TaskNodePath.getExecutionNodePath(jobName);
        List<String> items = zooKeeper.getChildren(executionNodePath);
        boolean isRunning = false;
        if (items != null) {
            for (String item : items) {
                String runningNodePath = TaskNodePath.getExecutionNodePath(jobName, item, "running");
                if (zooKeeper.checkExists(runningNodePath)) {
                    isRunning = true;
                    break;
                }
            }
        }
        String enabledNodePath = TaskNodePath.getConfigNodePath(jobName, "enabled");
        if (Boolean.parseBoolean(zooKeeper.getData(enabledNodePath))) {
            if (isRunning) {
                restApiJobInfo.setRunningStatus(TaskStatus.RUNNING.name());
            } else {
                restApiJobInfo.setRunningStatus(TaskStatus.READY.name());
            }
        } else {
            if (isRunning) {
                restApiJobInfo.setRunningStatus(TaskStatus.STOPPING.name());
            } else {
                restApiJobInfo.setRunningStatus(TaskStatus.STOPPED.name());
            }
        }
    }

    private void setJobConfig(ZooKeeper zooKeeper, RestApiJobInfo restApiJobInfo,
                              String jobName) {
        String configNodePath = TaskNodePath.getConfigNodePath(jobName);
        if (zooKeeper.checkExists(configNodePath)) {
            restApiJobInfo
                    .setDescription(zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "description")));
            restApiJobInfo.setEnabled(
                    Boolean.valueOf(zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "enabled"))));

            RestApiJobConfig restApiJobConfig = new RestApiJobConfig();
            restApiJobConfig
                    .setJobClass(zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "jobClass")));
            restApiJobConfig.setJobType(zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "jobType")));
            restApiJobConfig.setShardingTotalCount(Integer.valueOf(
                    zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "shardingTotalCount"))));
            restApiJobConfig.setShardingItemParameters(
                    zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "shardingItemParameters")));
            restApiJobConfig.setJobParameter(
                    zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "jobParameter")));
            String timeZone = zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "timeZone"));
            if (timeZone == null || timeZone.trim().length() == 0) {
                timeZone = SaturnConstants.TIME_ZONE_ID_DEFAULT;
            }
            restApiJobConfig.setTimeZone(timeZone);
            restApiJobConfig.setCron(zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "cron")));
            restApiJobConfig.setPausePeriodDate(
                    zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "pausePeriodDate")));
            restApiJobConfig.setPausePeriodTime(
                    zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "pausePeriodTime")));
            String timeout4AlarmSecondsStr = zooKeeper
                    .getData(TaskNodePath.getConfigNodePath(jobName, "timeout4AlarmSeconds"));
            if (timeout4AlarmSecondsStr != null) {
                restApiJobConfig.setTimeout4AlarmSeconds(Integer.valueOf(timeout4AlarmSecondsStr));
            } else {
                restApiJobConfig.setTimeout4AlarmSeconds(0);
            }
            restApiJobConfig.setTimeoutSeconds(Integer.valueOf(
                    zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "timeoutSeconds"))));
            restApiJobConfig.setLoadLevel(
                    Integer.valueOf(zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "loadLevel"))));


            restApiJobConfig.setEnabledReport(Boolean.valueOf(
                    zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "enabledReport"))));
            restApiJobConfig
                    .setPreferList(zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "preferList")));
            restApiJobConfig.setUseDispreferList(Boolean.valueOf(
                    zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "useDispreferList"))));
            restApiJobConfig.setShowNormalLog(Boolean.valueOf(
                    zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "showNormalLog"))));
            restApiJobConfig.setProcessCountInterValSeconds(Integer.valueOf(
                    zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "processCountIntervalSeconds"))));

            restApiJobInfo.setJobConfig(restApiJobConfig);
        }
    }

    private void setStatics(ZooKeeper zooKeeper,
                            RestApiJobStatistics restApiJobStatistics, String jobName) {
        setProcessCount(zooKeeper, restApiJobStatistics, jobName);
        setTimes(zooKeeper, restApiJobStatistics, jobName);
    }

    private void setProcessCount(ZooKeeper zooKeeper,
                                 RestApiJobStatistics restApiJobStatistics, String jobName) {
        String processCountPath = TaskNodePath.getProcessCountPath(jobName);
        String processCountStr = zooKeeper.getData(processCountPath);
        if (processCountStr != null) {
            try {
                restApiJobStatistics.setProcessCount(Long.valueOf(processCountStr));
            } catch (NumberFormatException e) {
                log.error(e.getMessage(), e);
            }
        }
        String errorCountPath = TaskNodePath.getErrorCountPath(jobName);
        String errorCountPathStr = zooKeeper.getData(errorCountPath);
        if (errorCountPathStr != null) {
            try {
                restApiJobStatistics.setProcessErrorCount(Long.valueOf(errorCountPathStr));
            } catch (NumberFormatException e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    private void setTimes(ZooKeeper zooKeeper,
                          RestApiJobStatistics restApiJobStatistics, String jobName) {
        String executionNodePath = TaskNodePath.getExecutionNodePath(jobName);
        List<String> items = zooKeeper.getChildren(executionNodePath);
        if (items == null) {
            return;
        }
        List<String> lastBeginTimeList = new ArrayList<>();
        List<String> lastCompleteTimeList = new ArrayList<>();
        List<String> nextFireTimeList = new ArrayList<>();
        int runningItemSize = 0;
        for (String item : items) {
            if (getRunningIP(item, jobName, zooKeeper) == null) {
                continue;
            }
            ++runningItemSize;
            String lastBeginTime = zooKeeper
                    .getData(TaskNodePath.getExecutionNodePath(jobName, item, "lastBeginTime"));
            if (null != lastBeginTime) {
                lastBeginTimeList.add(lastBeginTime);
            }
            String lastCompleteTime = zooKeeper
                    .getData(TaskNodePath.getExecutionNodePath(jobName, item, "lastCompleteTime"));
            if (null != lastCompleteTime) {
                boolean isItemCompleted = zooKeeper
                        .checkExists(TaskNodePath.getExecutionNodePath(jobName, item, "completed"));
                boolean isItemRunning = zooKeeper
                        .checkExists(TaskNodePath.getExecutionNodePath(jobName, item, "running"));
                if (isItemCompleted && !isItemRunning) { // 如果作业分片已执行完毕，则添加该完成时间到集合中进行排序
                    lastCompleteTimeList.add(lastCompleteTime);
                }
            }
            String nextFireTime = zooKeeper
                    .getData(TaskNodePath.getExecutionNodePath(jobName, item, "nextFireTime"));
            if (null != nextFireTime) {
                nextFireTimeList.add(nextFireTime);
            }
        }
        if (!CollectionUtils.isEmpty(lastBeginTimeList)) {
            Collections.sort(lastBeginTimeList);
            try {
                restApiJobStatistics.setLastBeginTime(Long.valueOf(lastBeginTimeList.get(0))); // 所有分片中最近最早的开始时间
            } catch (NumberFormatException e) {
                log.error(e.getMessage(), e);
            }
        }
        if (!CollectionUtils.isEmpty(lastCompleteTimeList)
                && lastCompleteTimeList.size() == runningItemSize) { // 所有分配都完成才显示最近最晚的完成时间
            Collections.sort(lastCompleteTimeList);
            try {
                restApiJobStatistics.setLastCompleteTime(
                        Long.valueOf(lastCompleteTimeList.get(lastCompleteTimeList.size() - 1))); // 所有分片中最近最晚的完成时间
            } catch (NumberFormatException e) {
                log.error(e.getMessage(), e);
            }
        }
        if (!CollectionUtils.isEmpty(nextFireTimeList)) {
            Collections.sort(nextFireTimeList);
            try {
                restApiJobStatistics.setNextFireTime(Long.valueOf(nextFireTimeList.get(0))); // 所有分片中下次最早的开始时间
            } catch (NumberFormatException e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    private String getRunningIP(String item, String jobName, ZooKeeper zooKeeper) {
        String runningIp = null;
        String serverNodePath = TaskNodePath.getExecutorsNodePath(jobName);
        if (!zooKeeper.checkExists(serverNodePath)) {
            return runningIp;
        }
        List<String> servers = zooKeeper.getChildren(serverNodePath);
        for (String server : servers) {
            String sharding = zooKeeper.getData(TaskNodePath.getExecutorsNodePath(jobName, server, "sharding"));
            String toFind = "";
            if (Strings.isNullOrEmpty(sharding)) {
                continue;
            }
            for (String itemKey : Splitter.on(',').split(sharding)) {
                if (item.equals(itemKey)) {
                    toFind = itemKey;
                    break;
                }
            }
            if (!Strings.isNullOrEmpty(toFind)) {
                runningIp = server;
                break;
            }
        }
        return runningIp;
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void enableJob(final String namespace, final String jobName) throws RuntimeException {
        ReuseUtils
                .reuse(namespace, jobName, zooKeeperService, zooKeeperRepository, new ReuseCallBackWithoutReturn() {
                    @Override
                    public void call(ZooKeeper zooKeeper)
                            throws RuntimeException {
                        TaskConfigurationDto jobConfig = taskConfigurationDtoService
                                .getTaskConfigurationDtoByNameSpaceAndTaskId(namespace, jobName);
                        if (jobConfig == null) {
                            throw new RuntimeException(
                                    "不能启用该作业（" + jobName + "），因为该作业不存在");
                        }
                        if (jobConfig.getEnabled()) {
                            throw new RuntimeException(
                                    "该作业（" + jobName + "）已经处于启用状态");
                        }

                        String enabledNodePath = TaskNodePath.getConfigNodePath(jobName, "enabled");
                        long ctime = zooKeeper.getCtime(enabledNodePath);
                        long mtime = zooKeeper.getMtime(enabledNodePath);
                        checkUpdateStatusToEnableAllowed(ctime, mtime);

                        jobConfig.setEnabled(Boolean.TRUE);
                        jobConfig.setLastUpdateTime(new Date());
                        jobConfig.setLastUpdateBy("");
                        taskConfigurationDtoService.updateByPrimaryKey(jobConfig);
                        zooKeeper.update(enabledNodePath, "true");
                    }
                });
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void disableJob(final String namespace, final String jobName) throws RuntimeException {
        ReuseUtils
                .reuse(namespace, jobName, zooKeeperService, zooKeeperRepository, new ReuseCallBackWithoutReturn() {
                    @Override
                    public void call(ZooKeeper zooKeeper)
                            throws RuntimeException {
                        TaskConfigurationDto jobConfig = taskConfigurationDtoService
                                .getTaskConfigurationDtoByNameSpaceAndTaskId(namespace, jobName);
                        if (jobConfig == null) {
                            throw new RuntimeException(
                                    "不能禁用该作业（" + jobName + "），因为该作业不存在");
                        }
                        if (!jobConfig.getEnabled()) {
                            throw new RuntimeException(
                                    "该作业（" + jobName + "）已经处于禁用状态");
                        }

                        String enabledNodePath = TaskNodePath.getConfigNodePath(jobName, "enabled");
                        long mtime = zooKeeper.getMtime(enabledNodePath);
                        checkUpdateStatusToDisableAllowed(mtime);

                        jobConfig.setEnabled(Boolean.FALSE);
                        jobConfig.setLastUpdateTime(new Date());
                        jobConfig.setLastUpdateBy("");
                        taskConfigurationDtoService.updateByPrimaryKey(jobConfig);
                        zooKeeper.update(enabledNodePath, "false");
                    }
                });
    }

    @Override
    public void updateJobCron(final String namespace, final String jobName, final String cron,
                              final Map<String, String> customContext) throws RuntimeException {
        ReuseUtils
                .reuse(namespace, jobName, zooKeeperService, zooKeeperRepository, new ReuseCallBackWithoutReturn() {
                    @Override
                    public void call(ZooKeeper zooKeeper)
                            throws RuntimeException {
                        String cronNodePath = TaskNodePath.getConfigNodePath(jobName, "cron");
                        long mtime = zooKeeper.getMtime(cronNodePath);
                        checkUpdateConfigAllowed(mtime);
                        String oldcronStr = zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "cron"));
                        if (!cron.equals(oldcronStr)) {
                            taskService.updateTaskCron(namespace, jobName, cron, customContext, "");
                        }
                    }
                });
    }

    private void checkUpdateConfigAllowed(long lastMtime) throws RuntimeException {
        if (Math.abs(System.currentTimeMillis() - lastMtime) < STATUS_UPDATE_FORBIDDEN_INTERVAL_IN_MILL_SECONDS) {
            String errMsg = "The update interval time cannot less than "
                    + STATUS_UPDATE_FORBIDDEN_INTERVAL_IN_MILL_SECONDS / 1000 + " seconds";
            log.warn(errMsg);
            throw new RuntimeException(errMsg);
        }
    }

    @Override
    public void runJobAtOnce(final String namespace, final String jobName, final Map<String, Object> triggeredData)
            throws RuntimeException {
        ReuseUtils
                .reuse(namespace, jobName, zooKeeperService, zooKeeperRepository, new ReuseCallBackWithoutReturn() {
                    @Override
                    public void call(ZooKeeper zooKeeper)
                            throws RuntimeException {
                        TaskStatus js = taskService.getTaskStatus(namespace, jobName);

                        if (!TaskStatus.READY.equals(js)) {
                            throw new RuntimeException(
                                    String.format(JOB_STATUS_NOT_CORRECT_TEMPATE, TaskStatus.READY.name()));
                        }
                        List<TaskExecutorStatus> jobServersStatus = taskService.getTaskExecutorStatusList(namespace, jobName);

                        if (CollectionUtils.isEmpty(jobServersStatus)) {
                            throw new RuntimeException(NO_EXECUTOR_FOUND);
                        }

                        boolean everExecute = false;
                        for (TaskExecutorStatus jobServerStatus : jobServersStatus) {
                            if (TaskExecutorStatus12.ONLINE.equals(jobServerStatus.getTaskExecutorStatus12())) {
                                everExecute = true;
                                String executorName = jobServerStatus.getExecutorName();
                                String path = TaskNodePath.getRunOneTimePath(jobName, executorName);
                                if (zooKeeper.checkExists(path)) {
                                    zooKeeper.delete(path);
                                }
                                String triggeredDataStr = gson.toJson(triggeredData);
                                zooKeeper.create(path, triggeredDataStr);
                                log.info("runAtOnce namespace:{}, jobName:{}, executorName:{}, triggeredData:{}",
                                        namespace, jobName, executorName, triggeredDataStr);
                            }
                        }

                        if (!everExecute) {
                            throw new RuntimeException(
                                    ALL_EXECUTORS_ARE_OFFLINE);
                        }
                    }

                });
    }

    @Override
    public void stopJobAtOnce(final String namespace, final String jobName) throws RuntimeException {
        ReuseUtils
                .reuse(namespace, jobName, zooKeeperService, zooKeeperRepository, new ReuseCallBackWithoutReturn() {
                    @Override
                    public void call(ZooKeeper zooKeeper)
                            throws RuntimeException {
                        // if job is already STOPPED then return
                        TaskStatus js = taskService.getTaskStatus(namespace, jobName);
                        if (TaskStatus.STOPPED.equals(js)) {
                            log.debug("job is already stopped");
                            return;
                        }

                        // For other Job types where the status is neither STOPPING nor STOPPED, cannot be stopped at
                        // once
                        if (TaskStatus.READY.equals(js) || TaskStatus.RUNNING.equals(js)) {
                            throw new RuntimeException(
                                    "job cannot be stopped while its status is READY or RUNNING");
                        }

                        List<String> jobServerList = taskService.getExecutorIdList(namespace, jobName);

                        if (CollectionUtils.isEmpty(jobServerList)) {
                            throw new RuntimeException(NO_EXECUTOR_FOUND);
                        }

                        for (String executorName : jobServerList) {
                            String path = TaskNodePath.getStopOneTimePath(jobName, executorName);
                            if (zooKeeper.checkExists(path)) {
                                zooKeeper.delete(path);
                            }
                            zooKeeper.create(path);
                            log.info("stopAtOnce namespace:{}, jobName:{}, executorName:{}", namespace, jobName,
                                    executorName);
                        }
                    }
                });
    }

    @Override
    public void deleteJob(final String namespace, final String jobName) throws RuntimeException {
        ReuseUtils
                .reuse(namespace, jobName, zooKeeperService, zooKeeperRepository, new ReuseCallBackWithoutReturn() {
                    @Override
                    public void call(ZooKeeper zooKeeper)
                            throws RuntimeException {
                        TaskStatus js = taskService.getTaskStatus(namespace, jobName);

                        if (!TaskStatus.STOPPED.equals(js)) {
                            throw new RuntimeException(
                                    String.format(JOB_STATUS_NOT_CORRECT_TEMPATE, TaskStatus.STOPPED.name()));
                        }

                        taskService.removeTask(namespace, jobName);
                        log.info("job:{} deletion done", jobName);
                    }
                });
    }

    @Override
    public void raiseAlarm(final String namespace, final String jobName, final String executorName,
                           final Integer shardItem, final AlarmInfo alarmInfo) throws RuntimeException {
        ReuseUtils
                .reuse(namespace, jobName, zooKeeperService, zooKeeperRepository, new ReuseCallBackWithoutReturn() {
                    @Override
                    public void call(ZooKeeper zooKeeper)
                            throws RuntimeException {
                        try {
                            // verify executor exist or not
                            if (!zooKeeper.checkExists(TaskNodePath.getExecutorsNodePath(jobName, executorName))) {
                                throw new RuntimeException(
                                        String.format("The executor {%s} does not exists.", executorName));
                            }

                            reportAlarmService.raise(namespace, jobName, executorName, shardItem, alarmInfo);
                        } catch (RuntimeException e) {
                            log.warn("ReportAlarmException: {}", e);
                            throw new RuntimeException(
                                    e.getMessage());
                        }
                    }
                });
    }

    @Override
    public void raiseExecutorRestartAlarm(final String namespace, final String executorName, final AlarmInfo alarmInfo)
            throws RuntimeException {
        ReuseUtils.reuse(namespace, zooKeeperService, zooKeeperRepository, new ReuseCallBackWithoutReturn() {
            @Override
            public void call(ZooKeeper zooKeeper) throws RuntimeException {
                try {
                    String restartTime = obtainRestartTime(alarmInfo);
                    reportAlarmService.executorRestart(namespace, executorName, restartTime);
                } catch (RuntimeException e) {
                    log.warn("ReportAlarmException: {}", e);
                    throw new RuntimeException(e.getMessage());
                }
            }
        });
    }

    private String obtainRestartTime(AlarmInfo alarmInfo) {
        String msg = alarmInfo.getMessage();
        int idx = msg.indexOf(EXECUTOR_RESTART_TIME_PREFIX);
        if (idx > 0) {
            return msg.substring(idx + EXECUTOR_RESTART_TIME_PREFIX.length());
        }

        return "";
    }

    private void checkUpdateStatusToEnableAllowed(long ctime, long mtime) throws RuntimeException {
        if (Math.abs(System.currentTimeMillis() - ctime)
                < OPERATION_FORBIDDEN_INTERVAL_AFTER_CREATION_IN_MILL_SECONDS) {
            String errMsg =
                    "Cannot enable the job until " + OPERATION_FORBIDDEN_INTERVAL_AFTER_CREATION_IN_MILL_SECONDS / 1000
                            + " seconds after job creation!";
            log.warn(errMsg);
            throw new RuntimeException(errMsg);
        }

        if (Math.abs(System.currentTimeMillis() - mtime) < STATUS_UPDATE_FORBIDDEN_INTERVAL_IN_MILL_SECONDS) {
            String errMsg = "The update interval time cannot less than "
                    + STATUS_UPDATE_FORBIDDEN_INTERVAL_IN_MILL_SECONDS / 1000 + " seconds";
            log.warn(errMsg);
            throw new RuntimeException(errMsg);
        }
    }

    private void checkUpdateStatusToDisableAllowed(long lastMtime) throws RuntimeException {
        if (Math.abs(System.currentTimeMillis() - lastMtime) < STATUS_UPDATE_FORBIDDEN_INTERVAL_IN_MILL_SECONDS) {
            String errMsg = "The update interval time cannot less than "
                    + STATUS_UPDATE_FORBIDDEN_INTERVAL_IN_MILL_SECONDS / 1000 + " seconds";
            log.warn(errMsg);
            throw new RuntimeException(errMsg);
        }
    }

    @Override
    public void updateJob(final String namespace, final String jobName, final TaskConfiguration taskConfiguration)
            throws RuntimeException {
        ReuseUtils
                .reuse(namespace, jobName, zooKeeperService, zooKeeperRepository, new ReuseCallBackWithoutReturn() {
                    @Override
                    public void call(ZooKeeper zooKeeper)
                            throws RuntimeException {
                        TaskStatus js = taskService.getTaskStatus(namespace, jobName);

                        if (!TaskStatus.STOPPED.equals(js)) {
                            throw new RuntimeException(
                                    String.format(JOB_STATUS_NOT_CORRECT_TEMPATE, TaskStatus.STOPPED.name()));
                        }
                        taskService.updateTaskConfiguration(namespace, taskConfiguration, "");
                        log.info("job {} update done", jobName);
                    }
                });
    }


    @Override
    public List<TaskConfiguration> getJobConfigList(String namespace) throws RuntimeException {
        return taskService.getTaskConfigurationList(namespace);
    }

}
